package com.atguigu.day02.test;

import com.sunfeng.TimeUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @Author : 孙峰
 * @Description:
 * @Date : 2022/6/30  9:57
 */
@Slf4j
public class ExecutorCompletionServiceTest01 {
    public static void main(String[] args) throws InterruptedException, ExecutionException {


    }

    /**
     * 案例一：向不同电商平台询价，并保存
     *  * 内部通过阻塞队列+FutureTask，实现了任务先完成可优先获取到，即结果按照完成的先后顺序排序。
     *  * 内部有一个先进先出的队列，用于保存已经执行完成的Future，通过调用它的take方法或者poll方法，可以获取到一个已经执行完成的Future，进而通过调用Future接口的实现类的get方法获取到最终结果。
     *  * <p>
     *  * Callable + Future 可以实现多个Task并行执行，但是遇到前面的task执行较慢时，需要阻塞等待前面的task执行完才能获取到后面task的执行结果。
     *  * CompletionService的主要功能就是一边生成任务，一边获取任务的返回值。让两件事分开执行，任务之间不会互相阻塞，可以实现执行完的先获取结果，不在依赖任务顺序。
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void test1() throws InterruptedException, ExecutionException {
        long start = System.currentTimeMillis();
        //创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);
        //创建 CompletionService
        ExecutorCompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
        //用于保存Future对象
        List<Future<Integer>> futures = new ArrayList<>();
        //提交异步任务，并保存 future到 futures
        futures.add(cs.submit(() -> getPriceByS1(5)));
        futures.add(cs.submit(() -> getPriceByS2(3)));
        futures.add(cs.submit(() -> getPriceByS3(2)));

        Integer result = null;

        try {
            for (int i = 0; i < 3; i++) {
                Future<Integer> take = cs.take();
                futures.remove(take);
                result = take.get();
                if (result != null) {
                    break;
                }
            }


        } catch (Exception e) {
            log.error("任务被取消");
        } finally {
            for (Future<Integer> future : futures) {
                // 取消剩下的任务
                future.cancel(true);
            }
        }
        executor.shutdown();
        while (true) {
            if (executor.isTerminated()) {
                long end = System.currentTimeMillis();
                System.out.println("耗时：" + (end - start) / 1000 + " s");
                break;
            }
        }
    }

    /**
     * 案例二：并行地调用多个服务，只要有一个成功就返回结果
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void test2() throws InterruptedException, ExecutionException {
        long start = System.currentTimeMillis();

        //创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);
        //创建 CompletionService
        ExecutorCompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
        //异步向电商S1询价
        cs.submit(() -> getPriceByS1(2));
        //异步向电商S2询价
        cs.submit(() -> getPriceByS2(8));
        //异步向电商S3询价
        cs.submit(() -> getPriceByS3(4));

        //将结果异步保存到数据库
        for (int i = 0; i < 3; i++) {
            Future<Integer> future = cs.take();
            Integer result = future.get();
            System.out.println("任务【" + result + "】完成");
            executor.execute(() -> {
                try {
                    // 这里也采用了异步提交
                    save(result);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }


        executor.shutdown();
        while (true) {
            if (executor.isTerminated()) {
                long end = System.currentTimeMillis();
                System.out.println("耗时：" + (end - start) / 1000 + " s");
                break;
            }
        }
    }

    private static Integer getPriceByS2(int i) throws InterruptedException {
        TimeUtils.sleep(i);
        return i;
    }

    private static Integer getPriceByS1(int i) throws InterruptedException {
        TimeUtils.sleep(i);
        return i;
    }

    private static Integer getPriceByS3(int i) throws InterruptedException {
        TimeUtils.sleep(i);
        return i;
    }

    private static void save(Integer price) throws InterruptedException {
        TimeUtils.sleep(1);
        System.out.println("保存价格：" + price);
    }


}
